package com.example.mq.mqclient;

import com.example.mq.common.dto.BasicReturns;
import com.example.mq.common.dto.Request;
import com.example.mq.common.dto.Response;
import com.example.mq.common.dto.SubScribeReturns;
import com.example.mq.common.exception.MqException;
import com.example.mq.common.tool.BinaryTool;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.net.Socket;
import java.net.SocketException;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class Connection {
    private final Socket socket;
    private final ConcurrentHashMap<String, Channel> channelMap = new ConcurrentHashMap<>(); // 用于存储多个 Channel 的映射关系

    private final InputStream inputStream;  // 输入流，用于从服务器读取数据
    private final OutputStream outputStream; // 输出流，用于向服务器发送数据
    private final DataInputStream dataInputStream; // 带有数据格式的输入流，用于读取服务器的响应
    private final DataOutputStream dataOutputStream; // 带有数据格式的输出流，用于发送请求数据

    private final ExecutorService callbackPool; // 线程池，用于处理回调

    // 构造函数，初始化连接和流对象
    public Connection(String host, int port) throws IOException {
        socket = new Socket(host, port); // 连接到指定的服务器
        inputStream = socket.getInputStream(); // 获取输入流
        outputStream = socket.getOutputStream(); // 获取输出流
        dataInputStream = new DataInputStream(inputStream); // 使用 DataInputStream 读取数据
        dataOutputStream = new DataOutputStream(outputStream); // 使用 DataOutputStream 发送数据

        // 初始化一个大小为4的线程池，处理异步回调任务
        callbackPool = Executors.newFixedThreadPool(4);

        // 创建一个线程，持续监听服务器的响应
        Thread t = new Thread(() -> {
            try {
                while (!socket.isClosed()) { // 当连接未关闭时，循环读取响应数据
                    Response response = readResponse(); // 读取服务器的响应
                    dispatchResponse(response); // 将响应数据分发到对应的 Channel 处理
                }
            } catch (SocketException e) {
                log.info("[Connection] 连接正常断开!"); // 处理连接正常关闭的情况
            } catch (IOException | ClassNotFoundException | MqException e) {
                log.error("[Connection] 连接异常断开!", e); // 处理连接异常断开的情况
            }
        });
        t.start(); // 启动监听线程
    }

    // 关闭连接，释放资源
    public void close() {
        try {
            callbackPool.shutdownNow(); // 关闭线程池
            channelMap.clear(); // 清空所有 Channel 的映射
            inputStream.close(); // 关闭输入流
            outputStream.close(); // 关闭输出流
            socket.close(); // 关闭 socket 连接
        } catch (IOException e) {
            log.error("关闭连接时发生异常!", e); // 处理关闭连接时的异常
        }
    }

    // 将响应分发到对应的 Channel，判断是控制响应还是推送的消息
    private void dispatchResponse(Response response) throws IOException, ClassNotFoundException, MqException {
        if (response.getType() == 0xc) { // 如果响应类型为 0xc，表示是服务器推送的消息
            SubScribeReturns subScribeReturns = (SubScribeReturns) BinaryTool.fromBytes(response.getPayload()); // 反序列化推送的消息
            Channel channel = channelMap.get(subScribeReturns.getChannelId()); // 根据 channelId 获取对应的 Channel
            if (channel == null) {
                throw new MqException("[Connection] 该消息对应的 channel 在客户端中不存在! channelId=" + subScribeReturns.getChannelId());
            }
            // 提交到线程池异步处理推送的消息
            callbackPool.submit(() -> {
                try {
                    channel.getConsumer().handleDelivery(subScribeReturns.getConsumerTag(), subScribeReturns.getBasicProperties(),
                            subScribeReturns.getBody()); // 调用回调函数处理消息
                } catch (MqException | IOException e) {
                    log.error("处理订阅消息时发生异常!", e);
                }
            });
        } else { // 处理正常的控制请求响应
            BasicReturns basicReturns = (BasicReturns) BinaryTool.fromBytes(response.getPayload()); // 反序列化控制响应
            Channel channel = channelMap.get(basicReturns.getChannelId()); // 获取接收响应的逻辑连接
            if (channel == null) {
                throw new MqException("[Connection] 该消息对应的 channel 在客户端中不存在! channelId=" + basicReturns.getChannelId());
            }
            channel.putReturns(basicReturns); // 将响应结果传递给对应的 Channel
        }
    }

    // 发送请求到服务器
    public void writeRequest(Request request) throws IOException {
        dataOutputStream.writeInt(request.getType()); // 发送请求类型
        dataOutputStream.writeInt(request.getLength()); // 发送请求长度
        dataOutputStream.write(request.getPayload()); // 发送请求的负载
        dataOutputStream.flush(); // 刷新输出流，确保数据发送出去
        log.debug("[Connection] 发送请求! type={}, length={}", request.getType(), request.getLength()); // 记录请求日志
    }

    // 从服务器读取响应
    public Response readResponse() throws IOException {
        Response response = new Response(); // 创建响应对象
        response.setType(dataInputStream.readInt()); // 读取响应类型
        response.setLength(dataInputStream.readInt()); // 读取响应长度
        byte[] payload = new byte[response.getLength()]; // 创建负载缓冲区
        int n = dataInputStream.read(payload); // 读取负载数据
        if (n != response.getLength()) { // 如果读取数据长度不匹配，抛出异常
            throw new IOException("读取的响应数据不完整!");
        }
        response.setPayload(payload); // 设置响应的负载
        log.debug("[Connection] 收到响应! type={}, length={}", response.getType(), response.getLength()); // 记录响应日志
        return response; // 返回响应对象
    }

    // 在当前 Connection 中创建一个新的 Channel
    public Channel createChannel() throws IOException {
        String channelId = "C-" + UUID.randomUUID().toString(); // 生成一个唯一的 Channel ID
        Channel channel = new Channel(channelId, this); // 创建新的 Channel 对象
        channelMap.put(channelId, channel); // 将 Channel 添加到映射表中
        boolean ok = channel.createChannel(); // 向服务器注册该 Channel
        if (!ok) {
            channelMap.remove(channelId); // 如果注册失败，删除该 Channel
            return null;
        }
        return channel;
    }
}
